package com.gaofans.nrpc.client.bootstrap;

import com.gaofans.nrpc.client.handler.ClientFilter;
import com.gaofans.nrpc.client.handler.ResponseEventProcessor;
import com.gaofans.nrpc.client.route.Router;
import com.gaofans.nrpc.client.route.ServiceEvent;
import com.gaofans.nrpc.client.route.impl.StaticRouter;
import com.gaofans.nrpc.client.route.impl.StaticRouterPredicate;
import com.gaofans.nrpc.common.context.SelectableExecutor;
import com.gaofans.nrpc.common.id.IdGenerator;
import com.gaofans.nrpc.common.processor.ServiceAuthority;
import com.gaofans.nrpc.common.processor.SimpleEventListener;
import com.gaofans.nrpc.common.processor.Sink;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author GaoFans
 */
public class SimpleChannelManager implements ChannelManager{

    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleChannelManager.class);

    private final Map<ServiceAuthority,Channel> CHANNELS = new ConcurrentHashMap<>();
    private final ChannelBootstrap bootstrap;
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final List<Router> routers;

    public SimpleChannelManager(ChannelBootstrap bootstrap) {
        this.bootstrap = bootstrap;
        routers = new CopyOnWriteArrayList<>();
        routers.add(new StaticRouter(0,new StaticRouterPredicate()));
        this.bootstrap.closeEventHandler().addListener(this.getClass().getSimpleName(), getListener());
    }


    @Override
    public synchronized Channel connect(ServiceAuthority info) {
        if (isClosed.get()){
            throw new IllegalStateException("closed");
        }
        Channel channel = CHANNELS.get(info);
        if (channel != null){
            return channel;
        }
        try {
            channel = this.bootstrap.bootstrap(info).sync().channel();
            CHANNELS.put(info,channel);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        LOGGER.debug("host:{} port:{} 连接成功",info.getHost(),info.getPort());
        return channel;
    }

    @Override
    public Channel connect(URI uri) {
        for (Router router : routers) {
            if (router.match(uri)){
                ServiceAuthority authority = router.find(uri);
                return connect(authority);
            }
        }
        LOGGER.error("找不到指定的服务:{}",uri);
        throw new IllegalStateException("找不到指定的服务:"+uri);
    }

    @Override
    public Channel get(ServiceAuthority info) {
        if (isClosed.get()){
            throw new IllegalStateException("closed");
        }
        return CHANNELS.get(info);
    }

    @Override
    public Channel get(URI uri) {
        for (Router router : routers) {
            if (router.match(uri)){
                ServiceAuthority authority = router.find(uri);
                return get(authority);
            }
        }
        return null;
    }

    @Override
    public void close(ServiceAuthority info) {
        if (isClosed.get()){
            LOGGER.info("channel manager已经关闭");
            return;
        }
        Channel remove = CHANNELS.remove(info);
        if (remove != null){
            remove.close().addListener(future -> {
                if (future.isSuccess()){
                    LOGGER.error("host:{},port:{} 已关闭",info.getHost(),info.getPort());
                }else {
                    LOGGER.error("host:{},port:{} 关闭失败",info.getHost(),info.getPort());
                }
            });
        }
    }

    @Override
    public ResponseEventProcessor processor() {
        return bootstrap.processor();
    }

    @Override
    public SelectableExecutor chooser() {
        return bootstrap.chooser();
    }

    @Override
    public void register(Router router) {
        router.addListener(this.getClass().getSimpleName(), getListener());
        this.routers.add(router);
        routers.sort(Comparator.comparingInt(Router::order));
    }

    @Override
    public void addFilter(ClientFilter filter) {
        bootstrap.addFilter(filter);
    }

    @Override
    public IdGenerator idGenerator() {
        return bootstrap.idGenerator();
    }

    @Override
    public void close() throws Exception {
        if (isClosed.getAndSet(true)){
            LOGGER.info("channel manager已经关闭");
            return;
        }
        for (Map.Entry<ServiceAuthority, Channel> entry : CHANNELS.entrySet()) {
            entry.getValue().close();
        }
        CHANNELS.clear();
        bootstrap.close();
    }
    private SimpleEventListener<ServiceEvent<ServiceAuthority>> getListener() {
        return new SimpleEventListener<>(new Sink<ServiceEvent<ServiceAuthority>>() {
            @Override
            public Sink<ServiceEvent<ServiceAuthority>> next(ServiceEvent<ServiceAuthority> event) {
                switch (event.getType()) {
                    case ADD:
                        connect(event.getData());
                        break;
                    case REMOVE:
                        CHANNELS.remove(event.getData());
                        LOGGER.debug("channel {} removed",event.getData());
                        break;
                    case CLOSE:
                        close(event.getData());
                        break;
                    default:
                        break;
                }
                return null;
            }

            @Override
            public void complete() {
            }

            @Override
            public void error(Throwable throwable) {
            }
        });
    }
}
